package com.hgx.flink.scala.demo

import java.util
import java.util.Properties

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests


object StreamKafka {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.enableCheckpointing(5000)

    // kafka 配置
    val ZOOKEEPER_HOST = "172.128.9.125:2181"
    val KAFKA_BROKERS = "172.128.9.125:9092"
    val TRANSACTION_GROUP = "car_hailing_gps_group01"
    val TOPIC_NAME = "topic.car_hailing_gps"
    val kafkaProps = new Properties()
    kafkaProps.setProperty("zookeeper.connect", ZOOKEEPER_HOST)
    kafkaProps.setProperty("bootstrap.servers", KAFKA_BROKERS)
    kafkaProps.setProperty("group.id", TRANSACTION_GROUP)

    val flinkKafkaConsumer = new FlinkKafkaConsumer[String](TOPIC_NAME, new SimpleStringSchema(), kafkaProps)
    val stream = env.addSource(flinkKafkaConsumer)

    stream.print()

    //    elasticsearch 写入
    val httpHosts = new util.ArrayList[HttpHost]
    httpHosts.add(new HttpHost("172.128.9.125", 9200, "http"))

    val esSinkBuilder = new ElasticsearchSink.Builder[String](httpHosts,
      new ElasticsearchSinkFunction[String] {
        def createIndexRequest(element: String): IndexRequest = {
          val json = element

          val jsonObj = JSON.parseObject(json);
          val licensePlateColorCode = jsonObj.getString("licensePlateColorCode")
          val numberPlate = jsonObj.getString("numberPlate")
          val positioningTime = jsonObj.getLong("positioningTime")

          val key = licensePlateColorCode + "|" + numberPlate + "|" + positioningTime
          val map = new util.HashMap[String, String]()

          map.put(key, json)
          Requests.indexRequest()
            .index("car_hailing_gps_001")
            .source(map)
        }

        override def process(element: String, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
          requestIndexer.add(createIndexRequest(element))
        }
      }
    )
    //批量请求的配置；这将指示接收器在每个元素之后发出请求，否则将对它们进行缓冲。
    esSinkBuilder.setBulkFlushMaxActions(1)

    stream.addSink(esSinkBuilder.build())

    env.execute()

  }


}
